package com.atbeijing.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.bean.visitor.VisitorStats;
import com.atbeijing.gmall.realtime.utils.ClickHouseUtil;
import com.atbeijing.gmall.realtime.utils.DateTimeUtil;
import com.atbeijing.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * 访客主题DWS
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(4);
        //TODO 2.设置检查点 略

        //TODO 3.声明消费主题以及消费者组
        String pageLogSourceTopic = "dwd_page_log";
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";
        String groupId = "visitor_stats_group";

        //读取三条流
        DataStreamSource<String> pageLogSource = env.addSource(MyKafkaUtil.getKafkaSource(pageLogSourceTopic, groupId));
        DataStreamSource<String> uniqueVisitSource = env.addSource(MyKafkaUtil.getKafkaSource(uniqueVisitSourceTopic, groupId));
        DataStreamSource<String> userJumpDetailSource = env.addSource(MyKafkaUtil.getKafkaSource(userJumpDetailSourceTopic, groupId));

        //TODO 结构转换
        //页面日志流,每条数据表示1次页面访问
        SingleOutputStreamOperator<VisitorStats> pageLogDS = pageLogSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        jsonObj.getJSONObject("common").getString("vc"),
                        jsonObj.getJSONObject("common").getString("ch"),
                        jsonObj.getJSONObject("common").getString("ar"),
                        jsonObj.getJSONObject("common").getString("is_new"),
                        0L,
                        1L,
                        0L,
                        0L,
                        jsonObj.getJSONObject("page").getLong("during_time"),
                        jsonObj.getLong("ts")
                );
                String lastPageId = jsonObj.getJSONObject("page").getString("last_page_id");
                if (lastPageId == null || lastPageId.length() == 0) {
                    visitorStats.setSv_ct(1L);
                }
                return visitorStats;
            }
        });

        //独立访客明细流 一条数据表示一个独立访客
        SingleOutputStreamOperator<VisitorStats> uniqueVisitDS = uniqueVisitSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        jsonObj.getJSONObject("common").getString("vc"),
                        jsonObj.getJSONObject("common").getString("ch"),
                        jsonObj.getJSONObject("common").getString("ar"),
                        jsonObj.getJSONObject("common").getString("is_new"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        jsonObj.getLong("ts")
                );
                return visitorStats;
            }
        });

        //用户跳出明细流
        SingleOutputStreamOperator<VisitorStats> userJumpDetailDS = userJumpDetailSource.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                JSONObject jsonObj = JSON.parseObject(value);
                VisitorStats visitorStats = new VisitorStats(
                        "",
                        "",
                        jsonObj.getJSONObject("common").getString("vc"),
                        jsonObj.getJSONObject("common").getString("ch"),
                        jsonObj.getJSONObject("common").getString("ar"),
                        jsonObj.getJSONObject("common").getString("is_new"),
                        0L,
                        0L,
                        0L,
                        1L,
                        0L,
                        jsonObj.getLong("ts")
                );
                return visitorStats;
            }
        });

        //TODO union三条流,类似mysql的union
        SingleOutputStreamOperator<VisitorStats> reduceDS = pageLogDS
                .union(uniqueVisitDS, userJumpDetailDS)
                //设置Watermark以及设置事件时间字段
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                                    @Override
                                    public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                //按照维度进行分组
                .keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                        return Tuple4.of(value.getVc(), value.getCh(), value.getAr(), value.getIs_new());
                    }
                })
                //开窗
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //对窗口中的数据进行聚合计算,reduce一般用在返回类型和进行聚合的数据类型相同时,他维护了个累加器
                .reduce(
                        new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
                                value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                                value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());
                                return value1;
                            }
                        },
                        //向reduce的最终聚合结果-->累加器 添加统计时间
                        new ProcessWindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void process(Tuple4<String, String, String, String> key, Context context, Iterable<VisitorStats> elements, Collector<VisitorStats> out) throws Exception {
                                String start = DateTimeUtil.toYMDhms(new Date(context.window().getStart()));
                                String end = DateTimeUtil.toYMDhms(new Date(context.window().getEnd()));
                                for (VisitorStats element : elements) {
                                    element.setStt(start);
                                    element.setEdt(end);
                                    out.collect(element);
                                }
                            }
                        }

                );

        reduceDS.print("访客主题>>>>>>>>>");

        //TODO 写入ClickHouse,表要先存在
        reduceDS.addSink(ClickHouseUtil.getJdbcSink("insert into visitor_stats_1116 values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
